import os
import sys

AbsolutePath = os.path.abspath(__file__)
SuperiorCatalogue = os.path.dirname(AbsolutePath)
BaseDir = os.path.dirname(SuperiorCatalogue)
sys.path.insert(0, SuperiorCatalogue)

import numpy as np
import struct
from CarApi.TestToolApi import Debug
from multiprocessing import Process, Queue as ProcQueue, Event, Lock
import multiprocessing.sharedctypes as mps
from typing import Optional
from ctypes import *
import subprocess as sb
import re
import time
import select
import asyncio
import traceback
import socket
import random


class MediaCaptureServer(Process):

    def __init__(self, logging=True, daemon=True):
        super(MediaCaptureServer, self).__init__(daemon=daemon)
        if logging:
            self.__mainDebugQueue = ProcQueue(10)
        else:
            self.__mainDebugQueue = None
        self.__mainDebug = Debug(self.__mainDebugQueue, True, "下位机系统-服务器模块-流媒体采集进程")
        self.__ErrorStatus = Event()
        self.__MainVideoEvent = Event()
        self.__SubVideoEvent = Event()
        self.__AudioEvent = Event()
        # 以下是共享内存
        self.__mainVideoLock = Lock()
        self.__subVideoLock = Lock()
        self.__audioLock = Lock()
        self.MAIN_VIDEO_STATUS = mps.Value(c_bool, False, lock=self.__mainVideoLock)
        self.MAIN_VIDEO_DATA = mps.Array(c_char, size_or_initializer=1024, lock=self.__mainVideoLock)
        self.SUB_VIDEO_STATUS = mps.Value(c_bool, False, lock=self.__subVideoLock)
        self.SUB_VIDEO_DATA = mps.Array(c_char, size_or_initializer=1024, lock=self.__subVideoLock)
        self.AUDIO_STATUS = mps.Value(c_bool, False, lock=self.__audioLock)
        self.AUDIO_DATA = mps.Array(c_char, size_or_initializer=1024, lock=self.__audioLock)

    async def __mainVideoGetTask(self, cam, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入主视频流采集器")
        videoStream = await asyncio.create_subprocess_shell(
            f"ffmpeg -f video4linux2 -r 30 -s 640x480 -i /dev/{cam} " +
            f"-b:v 500k -sc_threshold 499 -vprofile baseline -pix_fmt yuv420p -vcodec h264_omx -f h264 -",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
            shell=True
        )
        while not self.__ErrorStatus.is_set():
            try:
                rawData = await videoStream.stdout.readexactly(1024)
                if not self.__MainVideoEvent.is_set():
                    await saveBuffer.put(rawData)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[主视频流采集器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[主视频流采集器] 该进程发生错误, 退出!")
                break
        videoStream.terminate()
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[主视频流采集器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[主视频流采集器] 总进程遇到错误, 退出该进程!")

    async def __subVideoGetTask(self, cam, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入子视频流采集器")
        videoStream = await asyncio.create_subprocess_shell(
            f"ffmpeg -f video4linux2 -r 30 -s 640x480 -i /dev/{cam} " +
            f"-b:v 1000k -sc_threshold 499 -vprofile baseline -pix_fmt yuv420p -vcodec h264_omx -f h264 -",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
            shell=True
        )
        while not self.__ErrorStatus.is_set():
            try:
                rawData = await videoStream.stdout.readexactly(1024)
                if not self.__SubVideoEvent.is_set():
                    await saveBuffer.put(rawData)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[子视频流采集器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[子视频流采集器] 该进程发生错误, 退出!")
                break
        videoStream.terminate()
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[子视频流采集器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[子视频流采集器] 总进程遇到错误, 退出该进程!")

    async def __audioGetTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入音频流采集器")
        audioStream = await asyncio.create_subprocess_shell(
            f"ffmpeg -f alsa -i 'plughw:2,0' -ac 2 -ar 48000 -b:a 500k -acodec libmp3lame -f mp3 -",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
            shell=True
        )
        while not self.__ErrorStatus.is_set():
            try:
                rawData = await audioStream.stdout.readexactly(1024)
                if not self.__AudioEvent.is_set():
                    await saveBuffer.put(rawData)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[音频流采集器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[音频流采集器] 该进程发生错误, 退出!")
                break
        audioStream.terminate()
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[音频流采集器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[音频流采集器] 总进程遇到错误, 退出该进程!")

    async def __mainVideoPutTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入主视频流输出器")
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty() and not self.MAIN_VIDEO_STATUS.value:
                    rawData = await saveBuffer.get()
                    self.MAIN_VIDEO_DATA.value = rawData
                    self.MAIN_VIDEO_STATUS.value = True
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[主视频流输出器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[主视频流输出器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[主视频流输出器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[主视频流输出器] 总进程遇到错误, 退出该进程!")

    async def __subVideoPutTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入子视频流输出器")
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty() and not self.SUB_VIDEO_STATUS.value:
                    rawData = await saveBuffer.get()
                    self.SUB_VIDEO_DATA.value = rawData
                    self.SUB_VIDEO_STATUS.value = True
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[子视频流输出器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[子视频流输出器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[子视频流输出器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[子视频流输出器] 总进程遇到错误, 退出该进程!")

    async def __audioPutTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入音频流输出器")
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty() and not self.AUDIO_STATUS.value:
                    rawData = await saveBuffer.get()
                    self.AUDIO_DATA.value = rawData
                    self.AUDIO_STATUS.value = True
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[音频流输出器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[音频流输出器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[音频流输出器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[音频流输出器] 总进程遇到错误, 退出该进程!")

    def run(self) -> None:
        self.__mainDebug.info("正在初始化进程")

        async def main():

            camera, _ = await (await asyncio.create_subprocess_shell(
                f"ls -l /sys/class/video4linux/ | grep virtual",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
                shell=True
            )).communicate()
            cameraMain = camera.decode().split(' ')[-3]

            camera, _ = await (await asyncio.create_subprocess_shell(
                f"ls -l /dev/ | grep usb",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
                shell=True
            )).communicate()
            cameraSub = camera.decode().split(' ')[-1].strip("\n")

            mainVideoBuffer = asyncio.Queue(30)
            subVideoBuffer = asyncio.Queue(30)
            audioBuffer = asyncio.Queue(30)
            mainVideoGetTask = asyncio.create_task(self.__mainVideoGetTask(cameraMain, mainVideoBuffer))
            subVideoGetTask = asyncio.create_task(self.__subVideoGetTask(cameraSub, subVideoBuffer))
            audioGetTask = asyncio.create_task(self.__audioGetTask(audioBuffer))
            mainVideoPutTask = asyncio.create_task(self.__mainVideoPutTask(mainVideoBuffer))
            subVideoPutTask = asyncio.create_task(self.__subVideoPutTask(subVideoBuffer))
            audioVideoPutTask = asyncio.create_task(self.__audioPutTask(audioBuffer))

            await asyncio.gather(
                mainVideoGetTask, subVideoGetTask, audioGetTask,
                mainVideoPutTask, subVideoPutTask, audioVideoPutTask
            )

        self.__mainDebug.info("进程开始运行")
        asyncio.run(main())
        self.__ErrorStatus.set()
        self.__mainDebug.warn("总进程遇到错误, 退出该进程!")

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()

    def getDebugText(self):
        self.__mainDebug.getInfo()

    def isMainVideoOpened(self):
        return not self.__MainVideoEvent.is_set()

    def isSubVideoOpened(self):
        return not self.__SubVideoEvent.is_set()

    def isAudioOpened(self):
        return not self.__AudioEvent.is_set()

    def setMainVideo(self, status: bool):
        if status:
            self.__MainVideoEvent.clear()
        else:
            self.__MainVideoEvent.set()

    def setSubVideo(self, status: bool):
        if status:
            self.__SubVideoEvent.clear()
        else:
            self.__SubVideoEvent.set()

    def setAudio(self, status: bool):
        if status:
            self.__AudioEvent.clear()
        else:
            self.__AudioEvent.set()


class CenterConsumerServer(Process):

    def __init__(self,
                 mainVideoSentAddr: tuple,
                 subVideoSentAddr: tuple,
                 audioSentAddr: tuple,
                 baseDataSentAddr: tuple,
                 videoRecvAddr: tuple,
                 audioRecvAddr: tuple,
                 controlDataRecvAddr: tuple,
                 logging=True,
                 daemon=True):
        super(CenterConsumerServer, self).__init__(daemon=daemon)
        if logging:
            self.__mainDebugQueue = ProcQueue(10)
        else:
            self.__mainDebugQueue = None
        self.__mainDebug = Debug(self.__mainDebugQueue, True, "下位机系统-服务器模块-中央传输进程")
        self.__ErrorStatus = Event()
        # 以下是要发送的共享内存
        self.MAIN_VIDEO_STATUS: Optional[mps.Value] = None
        self.MAIN_VIDEO_DATA: Optional[mps.Array] = None
        self.SUB_VIDEO_STATUS: Optional[mps.Value] = None
        self.SUB_VIDEO_DATA: Optional[mps.Array] = None
        self.AUDIO_STATUS: Optional[mps.Value] = None
        self.AUDIO_DATA: Optional[mps.Array] = None
        # 以下是要接收的共享内存
        self.__recvVideoLock = Lock()
        self.__recvAudioLock = Lock()
        self.RECV_VIDEO_STATUS = mps.Value(c_bool, False, lock=self.__recvVideoLock)
        self.RECV_VIDEO_DATA = mps.Array(c_char, size_or_initializer=1024, lock=self.__recvVideoLock)
        self.RECV_AUDIO_STATUS = mps.Value(c_bool, False, lock=self.__recvAudioLock)
        self.RECV_AUDIO_DATA = mps.Array(c_char, size_or_initializer=1024, lock=self.__recvAudioLock)
        # 以下是发送地址
        self.mainVideoSentAddr = mainVideoSentAddr
        self.subVideoSentAddr = subVideoSentAddr
        self.audioSentAddr = audioSentAddr
        self.baseDataSentAddr = baseDataSentAddr
        # 以下是接收地址
        self.videoRecvAddr = videoRecvAddr
        self.audioRecvAddr = audioRecvAddr
        self.controlDataRecvAddr = controlDataRecvAddr

    async def __mainVideoSentTask(self, mainVideoSocket: socket.socket, sentAddress: tuple):
        self.__mainDebug.info("已进入主视频流发送器")
        while not self.__ErrorStatus.is_set():
            try:
                if self.MAIN_VIDEO_STATUS.value:
                    rawData = self.MAIN_VIDEO_DATA.value
                    self.MAIN_VIDEO_STATUS.value = False
                    await asyncio.sleep(0)
                    mainVideoSocket.sendto(rawData, sentAddress)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[主视频流发送器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[主视频流发送器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[主视频流发送器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[主视频流发送器] 总进程遇到错误, 退出该进程!")

    async def __subVideoSentTask(self, subVideoSocket: socket.socket, sentAddress: tuple):
        self.__mainDebug.info("已进入子视频流发送器")
        while not self.__ErrorStatus.is_set():
            try:
                if self.SUB_VIDEO_STATUS.value:
                    rawData = self.SUB_VIDEO_DATA.value
                    self.SUB_VIDEO_STATUS.value = False
                    await asyncio.sleep(0)
                    subVideoSocket.sendto(rawData, sentAddress)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[子视频流发送器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[子视频流发送器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[子视频流发送器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[子视频流发送器] 总进程遇到错误, 退出该进程!")

    async def __audioSentTask(self, audioSocket: socket.socket, sentAddress: tuple):
        self.__mainDebug.info("已进入音频流发送器")
        while not self.__ErrorStatus.is_set():
            try:
                if self.AUDIO_STATUS.value:
                    rawData = self.AUDIO_DATA.value
                    self.AUDIO_STATUS.value = False
                    await asyncio.sleep(0)
                    audioSocket.sendto(rawData, sentAddress)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[音频流发送器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[音频流发送器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[音频流发送器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[音频流发送器] 总进程遇到错误, 退出该进程!")

    async def __videoRecvGetTask(self, videoRecvSocket: socket.socket, videoRecvPoll: select.epoll,
                                 saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入视频流接收器")
        while not self.__ErrorStatus.is_set():
            try:
                result = videoRecvPoll.poll(0)
                if len(result) == 0:
                    await asyncio.sleep(0)
                else:
                    rawData, _ = videoRecvSocket.recvfrom(1024)
                    await saveBuffer.put(rawData)
            except Exception as e:
                self.__mainDebug.error(
                    f"[视频流接收器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[视频流接收器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[视频流接收器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[视频流接收器] 总进程遇到错误, 退出该进程!")

    async def __audioRecvGetTask(self, audioRecvSocket: socket.socket, audioRecvPoll: select.epoll,
                                 saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入音频流接收器")
        while not self.__ErrorStatus.is_set():
            try:
                result = audioRecvPoll.poll(0)
                if len(result) == 0:
                    await asyncio.sleep(0)
                else:
                    rawData, _ = audioRecvSocket.recvfrom(1024)
                    await saveBuffer.put(rawData)
            except Exception as e:
                self.__mainDebug.error(
                    f"[音频流接收器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[音频流接收器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[音频流接收器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[音频流接收器] 总进程遇到错误, 退出该进程!")

    async def __videoRecvPutTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入接收视频流输出器")
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty() and not self.RECV_VIDEO_STATUS.value:
                    rawData = await saveBuffer.get()
                    self.RECV_VIDEO_DATA.value = rawData
                    self.RECV_VIDEO_STATUS.value = True
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[接收视频流输出器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[接收视频流输出器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[接收视频流输出器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[接收视频流输出器] 总进程遇到错误, 退出该进程!")

    async def __audioRecvPutTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入接收音频流输出器")
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty() and not self.RECV_AUDIO_STATUS.value:
                    rawData = await saveBuffer.get()
                    self.RECV_AUDIO_DATA.value = rawData
                    # print(self.RECV_AUDIO_DATA.value)
                    self.RECV_AUDIO_STATUS.value = True
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[接收音频流输出器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[接收音频流输出器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[接收音频流输出器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[接收音频流输出器] 总进程遇到错误, 退出该进程!")

    async def __baseDataSentTask(self, baseDataSentSocket: socket.socket, saveBuffer: asyncio.Queue,
                                 sentAddress: tuple):
        self.__mainDebug.info("已进入状态数据发送器")
        package = struct.Struct('< f f B f f d')
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty():
                    rawData = await saveBuffer.get()
                    await asyncio.sleep(0)
                    rawData = list(rawData.values())
                    rawData = package.pack(*rawData)
                    baseDataSentSocket.sendto(rawData, sentAddress)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[状态数据发送器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[状态数据发送器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[状态数据发送器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[状态数据发送器] 总进程遇到错误, 退出该进程!")

    async def __controlDataRecvTask(self, controlDataRecvSocket: socket.socket, controlDataRecvPoll: select.epoll,
                                    saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入控制数据接收器")
        while not self.__ErrorStatus.is_set():
            try:
                result = controlDataRecvPoll.poll(0)
                if len(result) == 0:
                    await asyncio.sleep(0)
                else:
                    rawData, _ = controlDataRecvSocket.recvfrom(1024)
                    await saveBuffer.put(rawData)
            except Exception as e:
                self.__mainDebug.error(
                    f"[控制数据接收器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[控制数据接收器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[控制数据接收器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[控制数据接收器] 总进程遇到错误, 退出该进程!")

    async def __baseDataGetTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入状态数据获取器")
        sentData = {
            'bat': 1, 'speed': 0, 'signal': 5, 'lit': 10673, 'dim': 2659, 'time': time.time()
        }
        while not self.__ErrorStatus.is_set():
            try:
                sentData['bat'] = random.randint(1, 100)
                await asyncio.sleep(0.001)
                sentData['speed'] = random.randint(0, 9000)
                await asyncio.sleep(0.001)
                sentData['signal'] = random.randint(0, 5)
                await asyncio.sleep(0.001)
                sentData['lit'] = 10673
                await asyncio.sleep(0.001)
                sentData['dim'] = 2659
                await asyncio.sleep(0.001)
                sentData['time'] = time.time()
                await saveBuffer.put(sentData)
            except Exception as e:
                self.__mainDebug.error(
                    f"[状态数据获取器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[状态数据获取器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[状态数据获取器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[状态数据获取器] 总进程遇到错误, 退出该进程!")

    async def __controlDataOutTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入控制数据输出器")
        package = struct.Struct('< ? ? ? ? f')
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty():
                    rawData = await saveBuffer.get()
                    commandData = package.unpack(rawData)
                    if commandData[0] and not commandData[1]:
                        """
                        --- API前进操作 ---
                        """
                        self.__mainDebug.debug("[任务执行器] 进行下位机前进操作")
                    if commandData[1] and not commandData[0]:
                        """
                        --- API后退操作 ---
                        """
                        self.__mainDebug.debug("[任务执行器] 进行下位机后退操作")
                    if commandData[2] and not commandData[3]:
                        """
                        --- API左转操作 ---
                        """
                        self.__mainDebug.debug("[任务执行器] 进行下位机左转操作")
                    if commandData[3] and not commandData[2]:
                        """
                        --- API右转操作 ---
                        """
                        self.__mainDebug.debug("[任务执行器] 进行下位机右转操作")
                    if commandData[4]:
                        """
                        --- API设置速度 ---
                        """
                        self.__mainDebug.debug("[任务执行器] 设置下位机速度")
                    if commandData[0] == commandData[1] == commandData[2] == commandData[3]:
                        """
                        --- API停止移动 ---
                        """
                        self.__mainDebug.debug("[任务执行器] 进行下位机停止操作")
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[控制数据输出器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[控制数据输出器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[控制数据输出器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[控制数据输出器] 总进程遇到错误, 退出该进程!")

    def run(self) -> None:
        self.__mainDebug.info("正在初始化进程")
        asyncioLoop = asyncio.new_event_loop()

        mainVideoSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # mainVideoSentAddr = self.mainVideoSentAddr
        mainVideoSentTask = asyncioLoop.create_task(self.__mainVideoSentTask(mainVideoSocket, self.mainVideoSentAddr))

        subVideoSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # subVideoSentAddr = self.subVideoSentAddr
        subVideoSentTask = asyncioLoop.create_task(self.__subVideoSentTask(subVideoSocket, self.subVideoSentAddr))

        audioSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # audioSentAddr = self.audioSentAddr
        audioSentTask = asyncioLoop.create_task(self.__audioSentTask(audioSocket, self.audioSentAddr))

        videoRecvSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        videoRecvSocket.bind(self.videoRecvAddr)
        videoRecvPoll = select.epoll()
        videoRecvPoll.register(videoRecvSocket, select.EPOLLIN)
        videoRecvBuffer = asyncio.Queue(50, loop=asyncioLoop)
        videoRecvGetTask = asyncioLoop.create_task(
            self.__videoRecvGetTask(videoRecvSocket, videoRecvPoll, videoRecvBuffer))
        videoRecvPutTask = asyncioLoop.create_task(self.__videoRecvPutTask(videoRecvBuffer))

        audioRecvSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        audioRecvSocket.bind(self.audioRecvAddr)
        audioRecvPoll = select.epoll()
        audioRecvPoll.register(audioRecvSocket, select.EPOLLIN)
        audioRecvBuffer = asyncio.Queue(50, loop=asyncioLoop)
        audioRecvGetTask = asyncioLoop.create_task(
            self.__audioRecvGetTask(audioRecvSocket, audioRecvPoll, audioRecvBuffer))
        audioRecvPutTask = asyncioLoop.create_task(self.__audioRecvPutTask(audioRecvBuffer))

        baseDataSentSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # baseDataSentAddr = self.baseDataSentAddr
        baseDataBuffer = asyncio.Queue(50, loop=asyncioLoop)
        baseDataSentTask = asyncioLoop.create_task(
            self.__baseDataSentTask(baseDataSentSocket, baseDataBuffer, self.baseDataSentAddr))
        baseDataGetTask = asyncioLoop.create_task(self.__baseDataGetTask(baseDataBuffer))

        controlDataRecvSocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        controlDataRecvSocket.bind(self.controlDataRecvAddr)
        controlDataRecvPoll = select.epoll()
        controlDataRecvPoll.register(controlDataRecvSocket, select.EPOLLIN)
        controlDataRecvBuffer = asyncio.Queue(50, loop=asyncioLoop)
        controlDataRecvTask = asyncioLoop.create_task(
            self.__controlDataRecvTask(controlDataRecvSocket, controlDataRecvPoll, controlDataRecvBuffer))
        controlDataOutTask = asyncioLoop.create_task(self.__controlDataOutTask(controlDataRecvBuffer))

        async def main():
            await asyncio.gather(
                mainVideoSentTask, subVideoSentTask, audioSentTask, videoRecvGetTask, videoRecvPutTask,
                audioRecvGetTask,
                audioRecvPutTask, baseDataSentTask, baseDataGetTask, controlDataRecvTask, controlDataOutTask
            )

        self.__mainDebug.info("进程开始运行")
        asyncioLoop.run_until_complete(main())
        self.__ErrorStatus.set()
        self.__mainDebug.warn("总进程遇到错误，退出该进程!")

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()

    def getDebugText(self):
        self.__mainDebug.getInfo()


class AudioOutServer(Process):

    def __init__(self, logging=True, daemon=True):
        super(AudioOutServer, self).__init__(daemon=daemon)
        if logging:
            self.__mainDebugQueue = ProcQueue(10)
        else:
            self.__mainDebugQueue = None
        self.__mainDebug = Debug(self.__mainDebugQueue, True, "下位机系统-服务器模块-音频流输出进程")
        self.__ErrorStatus = Event()
        # 以下是要输出的共享内存
        self.RECV_AUDIO_STATUS: Optional[mps.Value] = None
        self.RECV_AUDIO_DATA: Optional[mps.Array] = None

    async def __audioOutRecvTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入音频流接收器")
        while not self.__ErrorStatus.is_set():
            try:
                if self.RECV_AUDIO_STATUS.value:
                    rawData = self.RECV_AUDIO_DATA.value
                    self.RECV_AUDIO_STATUS.value = False
                    # print(rawData)
                    await saveBuffer.put(rawData)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[音频流接收器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[音频流接收器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[音频流接收器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[音频流接收器] 总进程遇到错误, 退出该进程!")

    async def __audioOutPutTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入音频流解码输出器")
        audioStream = await asyncio.create_subprocess_shell(
            f"ffmpeg -f mp3 -i - -f wav - | aplay -",
            stdin=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
            shell=True
        )
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty():
                    rawData = await saveBuffer.get()
                    print(len(rawData))
                    audioStream.stdin.write(rawData)
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[音频流解码输出器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[音频流解码输出器] 该进程发生错误, 退出!")
                break
        audioStream.terminate()
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[音频流解码输出器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[音频流解码输出器] 总进程遇到错误, 退出该进程!")

    def run(self) -> None:
        self.__mainDebug.info("正在初始化进程")

        async def main():
            audioOutBuffer = asyncio.Queue(50)
            audioOutRecvTask = asyncio.create_task(self.__audioOutRecvTask(audioOutBuffer))
            audioOutPutTask = asyncio.create_task(self.__audioOutPutTask(audioOutBuffer))
            await asyncio.gather(
                audioOutRecvTask, audioOutPutTask
            )

        self.__mainDebug.info("进程开始运行")
        asyncio.run(main())
        self.__ErrorStatus.set()
        self.__mainDebug.warn("总进程遇到错误，退出该进程!")

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()

    def getDebugText(self):
        self.__mainDebug.getInfo()


class VideoOutServer(Process):

    def __init__(self, logging=True, daemon=True):
        super(VideoOutServer, self).__init__(daemon=daemon)
        if logging:
            self.__mainDebugQueue = ProcQueue(10)
        else:
            self.__mainDebugQueue = None
        self.__mainDebug = Debug(self.__mainDebugQueue, True, "下位机系统-服务器模块-视频流输出进程")
        self.__ErrorStatus = Event()
        self.__outVideoLock = Lock()
        # 以下是要输出的共享内存
        self.RECV_VIDEO_STATUS: Optional[mps.Value] = None
        self.RECV_VIDEO_DATA: Optional[mps.Array] = None
        # 以下是要编码的共享内存
        self.OUT_VIDEO_STATUS = mps.Value(c_bool, False, lock=self.__outVideoLock)
        self.OUT_VIDEO_DATA = mps.copy(np.ctypeslib.as_ctypes(np.zeros((480, 640, 3), dtype=np.uint8)))

    async def __videoOutRecvTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入视频流接收器")
        while not self.__ErrorStatus.is_set():
            try:
                if self.RECV_VIDEO_STATUS.value:
                    rawData = self.RECV_VIDEO_DATA.value
                    self.RECV_VIDEO_STATUS.value = False
                    await saveBuffer.put(rawData)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[视频流接收器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[视频流接收器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[视频流接收器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[视频流接收器] 总进程遇到错误, 退出该进程!")

    async def __videoOutDecodeTask(self, saveInBuffer: asyncio.Queue, videoStream: asyncio.subprocess.Process):
        self.__mainDebug.info("已进入视频流解码器")
        while not self.__ErrorStatus.is_set():
            try:
                if not saveInBuffer.empty():
                    rawDataIn = await saveInBuffer.get()
                    videoStream.stdin.write(rawDataIn)
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[视频流解码器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[视频流解码器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[视频流解码器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[视频流解码器] 总进程遇到错误, 退出该进程!")

    async def __videoOutPutTask(self, saveOutBuffer: asyncio.Queue, videoStream: asyncio.subprocess.Process):
        self.__mainDebug.info("已进入视频流生成器")
        while not self.__ErrorStatus.is_set():
            try:
                rawDataOut = await videoStream.stdout.readexactly(480 * 640 * 3)
                await saveOutBuffer.put(rawDataOut)
            except Exception as e:
                self.__mainDebug.error(
                    f"[视频流生成器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[视频流生成器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[视频流生成器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[视频流生成器] 总进程遇到错误, 退出该进程!")

    async def __videoOutFreshTask(self, saveBuffer: asyncio.Queue):
        self.__mainDebug.info("已进入图像更新器")
        while not self.__ErrorStatus.is_set():
            try:
                if not saveBuffer.empty() and not self.OUT_VIDEO_STATUS.value:
                    rawData = await saveBuffer.get()
                    frame = np.frombuffer(rawData, np.uint8).reshape([480, 640, 3])
                    self.OUT_VIDEO_DATA[:] = frame
                    self.OUT_VIDEO_STATUS.value = True
                    await asyncio.sleep(0)
                else:
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainDebug.error(
                    f"[图像更新器] \n{traceback.format_exc()}\n----------Error----------"
                )
                self.__mainDebug.error("[图像更新器] 该进程发生错误, 退出!")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.error("[图像更新器] 出现全局错误申请, 退出该进程!")
        self.__ErrorStatus.set()
        self.__mainDebug.warn("[图像更新器] 总进程遇到错误, 退出该进程!")

    def run(self) -> None:
        self.__mainDebug.info("正在初始化进程")

        async def main():
            videoInBuffer = asyncio.Queue(50)
            videoOutBuffer = asyncio.Queue(50)
            videoStream = await asyncio.create_subprocess_shell(
                f"ffmpeg -f h264 -i - -f rawvideo -vcodec h264_mmal -pix_fmt bgr24 -",
                stdin=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
                stdout=asyncio.subprocess.PIPE,
                shell=True
            )

            videoOutRecvTask = asyncio.create_task(self.__videoOutRecvTask(videoInBuffer))
            videoOutDecodeTask = asyncio.create_task(self.__videoOutDecodeTask(videoInBuffer, videoStream))
            videoOutPutTask = asyncio.create_task(self.__videoOutPutTask(videoOutBuffer, videoStream))
            videoOutFreshTask = asyncio.create_task(self.__videoOutFreshTask(videoOutBuffer))

            await asyncio.gather(
                videoOutRecvTask, videoOutDecodeTask, videoOutPutTask, videoOutFreshTask
            )

        self.__mainDebug.info("进程开始运行")
        asyncio.run(main())
        self.__ErrorStatus.set()
        self.__mainDebug.warn("总进程遇到错误，退出该进程!")

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()

    def getDebugText(self):
        self.__mainDebug.getInfo()


if __name__ == '__main__':
    SENT_IP = '139.155.181.21'
    RECV_IP = '0.0.0.0'

    mediaCaptureServer = MediaCaptureServer()
    mediaCaptureServer.setMainVideo(True)
    mediaCaptureServer.setSubVideo(False)
    mediaCaptureServer.setAudio(False)

    MAIN_VIDEO_STATUS = mediaCaptureServer.MAIN_VIDEO_STATUS
    MAIN_VIDEO_DATA = mediaCaptureServer.MAIN_VIDEO_DATA
    SUB_VIDEO_STATUS = mediaCaptureServer.SUB_VIDEO_STATUS
    SUB_VIDEO_DATA = mediaCaptureServer.SUB_VIDEO_DATA
    AUDIO_STATUS = mediaCaptureServer.AUDIO_STATUS
    AUDIO_DATA = mediaCaptureServer.AUDIO_DATA

    centerConsumerServer = CenterConsumerServer(
        mainVideoSentAddr=(SENT_IP, 8000),
        subVideoSentAddr=(SENT_IP, 8001),
        audioSentAddr=(SENT_IP, 8002),
        baseDataSentAddr=(SENT_IP, 8003),
        videoRecvAddr=(RECV_IP, 6000),
        audioRecvAddr=(RECV_IP, 6001),
        controlDataRecvAddr=(RECV_IP, 6002)
    )

    centerConsumerServer.MAIN_VIDEO_STATUS = MAIN_VIDEO_STATUS
    centerConsumerServer.MAIN_VIDEO_DATA = MAIN_VIDEO_DATA
    centerConsumerServer.SUB_VIDEO_STATUS = SUB_VIDEO_STATUS
    centerConsumerServer.SUB_VIDEO_DATA = SUB_VIDEO_DATA
    centerConsumerServer.AUDIO_STATUS = AUDIO_STATUS
    centerConsumerServer.AUDIO_DATA = AUDIO_DATA

    RECV_VIDEO_STATUS = centerConsumerServer.RECV_VIDEO_STATUS
    RECV_VIDEO_DATA = centerConsumerServer.RECV_VIDEO_DATA
    RECV_AUDIO_STATUS = centerConsumerServer.RECV_AUDIO_STATUS
    RECV_AUDIO_DATA = centerConsumerServer.RECV_AUDIO_DATA

    videoOutServer = VideoOutServer()
    videoOutServer.RECV_VIDEO_STATUS = RECV_VIDEO_STATUS
    videoOutServer.RECV_VIDEO_DATA = RECV_VIDEO_DATA

    OUT_VIDEO_STATUS = videoOutServer.OUT_VIDEO_STATUS
    OUT_VIDEO_DATA = videoOutServer.OUT_VIDEO_DATA

    audioOutServer = AudioOutServer()
    audioOutServer.RECV_AUDIO_STATUS = RECV_AUDIO_STATUS
    audioOutServer.RECV_AUDIO_DATA = RECV_AUDIO_DATA

    mediaCaptureServer.openServer()
    centerConsumerServer.openServer()
    videoOutServer.openServer()
    audioOutServer.openServer()

    mediaCaptureServer.join()
    centerConsumerServer.join()
    videoOutServer.join()
    audioOutServer.join()
